package com.wuhanpe.leafid.generator

import com.mybatisflex.core.query.QueryWrapper
import com.wuhanpe.leafid.entity.LeafAlloc
import com.wuhanpe.leafid.entity.Result
import com.wuhanpe.leafid.factory.UpdateThreadFactory
import com.wuhanpe.leafid.mapper.LeafAllocMapper
import com.wuhanpe.leafid.model.Segment
import com.wuhanpe.leafid.model.SegmentBuffer
import com.wuhanpe.leafid.model.Status
import com.wuhanpe.qs.ultra.config.Slf4j
import com.wuhanpe.qs.ultra.config.Slf4j.Companion.log
import org.perf4j.StopWatch
import org.perf4j.slf4j.Slf4JStopWatch
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Component
import java.util.concurrent.*
import java.util.concurrent.atomic.AtomicLong
import kotlin.concurrent.Volatile

/**
 * @author zooy
 * @since 2025/9/12-16:18
 */
@Component
@Slf4j
class SegmentIDGenImpl: IDGen {

    companion object {
        /**
         * IDCache未初始化成功时的异常码
         */
        const val EXCEPTION_ID_IDCACHE_INIT_FALSE: Long = -1
        /**
         * key不存在时的异常码
         */
        const val EXCEPTION_ID_KEY_NOT_EXISTS: Long = -2
        /**
         * SegmentBuffer中的两个Segment均未从DB中装载时的异常码
         */
        const val EXCEPTION_ID_TWO_SEGMENTS_ARE_NULL: Long = -3
        /**
         * 最大步长不超过100,0000
         */
        const val MAX_STEP: Int = 1000000
        /**
         * 一个Segment维持时间为15分钟
         */
        const val SEGMENT_DURATION: Long = 15 * 60 * 1000L
    }

    @Volatile
    var initOK: Boolean = false

    var cache: MutableMap<String?, SegmentBuffer> = ConcurrentHashMap<String?, SegmentBuffer>()

    @Autowired
    lateinit var leafAllocMapper: LeafAllocMapper

    private val service: ExecutorService = ThreadPoolExecutor(
        5,
        Int.Companion.MAX_VALUE,
        60L,
        TimeUnit.SECONDS,
        SynchronousQueue<Runnable?>(),
        UpdateThreadFactory()
    )

    override fun get(key: String): Result {
        if (!initOK) {
            return Result(EXCEPTION_ID_IDCACHE_INIT_FALSE, Status.EXCEPTION)
        }
        if (cache.containsKey(key)) {
            val buffer: SegmentBuffer = cache.get(key)!!
            if (!buffer.initOk) {
                synchronized(buffer) {
                    if (!buffer.initOk) {
                        try {
                            updateSegmentFromDb(key, buffer.getCurrent())
                            log.info(
                                "Init buffer. Update leafkey {} {} from db",
                                key,
                                buffer.getCurrent()
                            )
                            buffer.initOk = true
                        } catch (e: java.lang.Exception) {
                            log.warn("Init buffer {} exception", buffer.getCurrent(), e)
                        }
                    }
                }
            }
            return getIdFromSegmentBuffer(cache.get(key)!!)
        }
        return Result(SegmentIDGenImpl.EXCEPTION_ID_KEY_NOT_EXISTS, Status.EXCEPTION)
    }

    override fun init(): Boolean {
        log.info("Init ...")

        // 确保加载到kv后才初始化成功
        updateCacheFromDb()
        initOK = true
        updateCacheFromDbAtEveryMinute()
        return initOK
    }

    private fun updateCacheFromDb() {
        log.info("update cache from db")
        val sw: StopWatch = Slf4JStopWatch()
        try {
            val dbTags: List<String?> = leafAllocMapper.selectListByQuery(QueryWrapper.create()).map { it.bizTag }
            if (dbTags == null || dbTags.isEmpty()) {
                return
            }
            val cacheTags: MutableList<String?> = ArrayList(cache.keys)
            val insertTagsSet: MutableSet<String?> = HashSet(dbTags)
            val removeTagsSet: MutableSet<String?> = HashSet(cacheTags)
            //db中新加的tags灌进cache
            for (i in cacheTags.indices) {
                val tmp = cacheTags.get(i)
                if (insertTagsSet.contains(tmp)) {
                    insertTagsSet.remove(tmp)
                }
            }
            for (tag in insertTagsSet) {
                val buffer: SegmentBuffer = SegmentBuffer()
                buffer.key = tag
                val segment: Segment = buffer.getCurrent()
                segment.value = AtomicLong(0)
                segment.max = 0
                segment.step = 0
                cache.put(tag, buffer)
                log.info("Add tag {} from db to IdCache, SegmentBuffer {}", tag, buffer)
            }
            //cache中已失效的tags从cache删除
            for (i in dbTags.indices) {
                val tmp = dbTags.get(i)
                if (removeTagsSet.contains(tmp)) {
                    removeTagsSet.remove(tmp)
                }
            }
            for (tag in removeTagsSet) {
                cache.remove(tag)
                log.info("Remove tag {} from IdCache", tag)
            }
        } catch (e: Exception) {
            log.warn("update cache from db exception", e)
        } finally {
            sw.stop("updateCacheFromDb")
        }
    }

    fun updateSegmentFromDb(key: String, segment: Segment) {
        val sw: StopWatch = Slf4JStopWatch()
        val buffer: SegmentBuffer = segment.buffer
        val leafAlloc: LeafAlloc
        if (!buffer.initOk) {
            leafAllocMapper.updateMaxId(key)
            leafAlloc = leafAllocMapper.selectOneByQuery(QueryWrapper.create().eq(LeafAlloc::bizTag, key))
            buffer.step = leafAlloc.step
            buffer.minStep = leafAlloc.step //leafAlloc中的step为DB中的step
        } else if (buffer.updateTimestamp == 0L) {
            leafAllocMapper.updateMaxId(key)
            leafAlloc = leafAllocMapper.selectOneByQuery(QueryWrapper.create().eq(LeafAlloc::bizTag, key))
            buffer.updateTimestamp = System.currentTimeMillis()
            buffer.step = leafAlloc.step
            buffer.minStep = leafAlloc.step //leafAlloc中的step为DB中的step
        } else {
            val duration: Long = System.currentTimeMillis() - buffer.updateTimestamp
            var nextStep: Int = buffer.step
            if (duration < SEGMENT_DURATION) {
                if (nextStep * 2 > MAX_STEP) {
                    //do nothing
                } else {
                    nextStep = nextStep * 2
                }
            } else if (duration < SEGMENT_DURATION * 2) {
                //do nothing with nextStep
            } else {
                nextStep = if (nextStep / 2 >= buffer.minStep) nextStep / 2 else nextStep
            }
            log.info(
                "leafKey[{}], step[{}], duration[{}mins], nextStep[{}]",
                key,
                buffer.step,
                String.format("%.2f", (duration.toDouble() / (1000 * 60))),
                nextStep
            )
            val temp: LeafAlloc = LeafAlloc()
            temp.bizTag = key
            temp.step = nextStep
            leafAllocMapper.updateMaxIdByCustomStep(temp)
            leafAlloc = leafAllocMapper.selectOneByQuery(QueryWrapper.create().eq(LeafAlloc::bizTag, key))
            buffer.updateTimestamp = System.currentTimeMillis()
            buffer.step = nextStep
            //leafAlloc的step为DB中的step
            buffer.minStep = leafAlloc.step
        }
        // must set value before set max
        val value: Long = leafAlloc.maxId - buffer.step
        segment.value.set(value)
        segment.max = leafAlloc.maxId
        segment.step = buffer.step
        sw.stop("updateSegmentFromDb", key + " " + segment)
    }

    fun getIdFromSegmentBuffer(buffer: SegmentBuffer): Result {
        while (true) {
            buffer.rLock().lock()
            try {
                val segment = buffer.getCurrent()
                if (!buffer.nextReady && (segment.getIdle() < 0.9 * segment.step) && buffer.threadRunning
                        .compareAndSet(false, true)
                ) {
                    service.execute(object : Runnable {
                        override fun run() {
                            val next: Segment = buffer.segments[buffer.nextPos()]
                            var updateOk = false
                            try {
                                updateSegmentFromDb(buffer.key!!, next)
                                updateOk = true
                                log.info("update segment {} from db {}", buffer.key, next)
                            } catch (e: java.lang.Exception) {
                                log.warn(buffer.key + " updateSegmentFromDb exception", e)
                            } finally {
                                if (updateOk) {
                                    buffer.wLock().lock()
                                    buffer.nextReady = true
                                    buffer.threadRunning.set(false)
                                    buffer.wLock().unlock()
                                } else {
                                    buffer.threadRunning.set(false)
                                }
                            }
                        }
                    })
                }
                val value: Long = segment.value.getAndIncrement()
                if (value < segment.max) {
                    return Result(value, Status.SUCCESS)
                }
            } finally {
                buffer.rLock().unlock()
            }
            waitAndSleep(buffer)
            buffer.wLock().lock()
            try {
                val segment = buffer.getCurrent()
                val value: Long = segment.value.getAndIncrement()
                if (value < segment.max) {
                    return Result(value, Status.SUCCESS)
                }
                if (buffer.nextReady) {
                    buffer.switchPos()
                    buffer.nextReady = false
                } else {
                    log.error("Both two segments in {} are not ready!", buffer)
                    return Result(EXCEPTION_ID_TWO_SEGMENTS_ARE_NULL, Status.EXCEPTION)
                }
            } finally {
                buffer.wLock().unlock()
            }
        }
    }

    private fun waitAndSleep(buffer: SegmentBuffer) {
        var roll = 0
        while (buffer.threadRunning.get()) {
            roll += 1
            if (roll > 10000) {
                try {
                    TimeUnit.MILLISECONDS.sleep(10)
                    break
                } catch (e: InterruptedException) {
                    log.warn("Thread {} Interrupted",Thread.currentThread().getName())
                    break
                }
            }
        }
    }

    private fun updateCacheFromDbAtEveryMinute() {
        val service = Executors.newSingleThreadScheduledExecutor(object : ThreadFactory {
            override fun newThread(r: Runnable?): Thread {
                val t = Thread(r)
                t.setName("check-idCache-thread")
                t.setDaemon(true)
                return t
            }
        })
        service.scheduleWithFixedDelay(object : Runnable {
            override fun run() {
                updateCacheFromDb()
            }
        }, 60, 60, TimeUnit.SECONDS)
    }
}
